package com.ningyue.water.mqttProcessor;

import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.*;
import org.springframework.stereotype.Component;

import java.net.URISyntaxException;

/**
 * Created by steven on 15-11-17.
 */
@Component
public class ApolloMqttProcessor {

    private final static String CONNECTION_STRING = "tcp://fddiaoyu.com:1883";
    private final static boolean CLEAN_START = true;
    private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s
    public  static Topic[] topics = {
            new Topic("china/beijing", QoS.EXACTLY_ONCE),
            new Topic("china/tianjin", QoS.AT_LEAST_ONCE),
            new Topic("china/henan", QoS.AT_MOST_ONCE)};
    public final  static long RECONNECTION_ATTEMPT_MAX=6;
    public final  static long RECONNECTION_DELAY=2000;

    public final static int SEND_BUFFER_SIZE=2*1024*1024;//发送最大缓冲为2M
    private MQTT mqtt;
    private CallbackConnection connection;

    public void connect() throws URISyntaxException {
        mqtt=new MQTT();
        //设置mqtt broker的ip和端口
        mqtt.setHost(CONNECTION_STRING);
        //连接前清空会话信息
        mqtt.setCleanSession(CLEAN_START);
        //设置重新连接的次数
        mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);
        //设置重连的间隔时间
        mqtt.setReconnectDelay(RECONNECTION_DELAY);
        //设置心跳时间
        mqtt.setKeepAlive(KEEP_ALIVE);
        //设置缓冲的大小
        mqtt.setSendBufferSize(SEND_BUFFER_SIZE);

        mqtt.setUserName("admin");

        mqtt.setPassword("password");

        connection = mqtt.callbackConnection();
        //添加
        // 连接的监听事件
        connection.listener(new Listener() {

            public void onDisconnected() {
                System.out.println("the connection disconnected!");
            }

            public void onConnected() {
                System.out.println("connect Broker successful");
            }

            public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) {
                // You can now process a received message from a topic.
                // Once process execute the ack runnable.
                ack.run();
                System.out.println("topic" + topic.toString() + "=" + new String(payload.getData()));
            }
            public void onFailure(Throwable value) {

            }




        });

        //添加连接事件
        connection.connect(new Callback<Void>() {
            /**
             * 连接失败的操作
             */
            public void onFailure(Throwable value) {
                // If we could not connect to the server.
                System.out.println("MQTTCallbackServer.CallbackConnection.connect.onFailure" + "连接失败......" + value.getMessage());
                value.printStackTrace();
            }

            /**
             * 连接成功的操作
             *
             * @param v
             */
            public void onSuccess(Void v) {

                Topic[] subTopics = {
                        new Topic("nanjing", QoS.EXACTLY_ONCE)};
                connection.subscribe(subTopics, new Callback<byte[]>() {
                    @Override
                    public void onSuccess(byte[] qoses) {
                        System.out.println("subscribe success fully!");
                    }

                    @Override
                    public void onFailure(Throwable throwable) {

                    }


                });

            }
        });



    }
    public void init() throws URISyntaxException {
        connect();
    }

    public void sendMessage(String message){

        final String topic = "nanjing";
        System.out.println("MQTTCallbackServer  publish  topic=" + topic + " message :" + message);
        connection.publish(topic, message.getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
            public void onSuccess(Void v) {
                // the pubish operation completed successfully.
                System.out.print("publish success");
            }

            public void onFailure(Throwable value) {
                value.printStackTrace();
            }
        });
    }
}
